package mq

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/nsqio/go-nsq"
)

// NSQPlugin 实现基于 NSQ 的消息队列
type NSQPlugin struct {
	nsqdAddress     string
	closed          bool
	startTime       int64
	errorCount      int
	lastError       string
	queueStats      map[string]*QueueStats
	statsMutex      sync.RWMutex
	consumersMutex  sync.Mutex
	activeConsumers map[string]*nsq.Consumer
}

// 确保 NSQPlugin 实现了 MQPlugin 接口
var _ MQPlugin = (*NSQPlugin)(nil)

// newNSQPlugin 创建一个新的 NSQPlugin 实例
func newNSQPlugin(config map[string]interface{}) (MQPlugin, error) {
	// 从配置中获取 NSQ 服务器地址
	nsqdAddress, ok := config["nsqd_address"].(string)
	if !ok || nsqdAddress == "" {
		return nil, fmt.Errorf("NSQ 服务器地址配置缺失或无效")
	}

	return &NSQPlugin{
		nsqdAddress:     nsqdAddress,
		closed:          false,
		startTime:       time.Now().Unix(),
		queueStats:      make(map[string]*QueueStats),
		activeConsumers: make(map[string]*nsq.Consumer),
	}, nil
}

// Type 返回消息队列类型
func (p *NSQPlugin) Type() MQType {
	return NSQ
}

// Publish 发布消息到指定主题
func (p *NSQPlugin) Publish(ctx context.Context, topic string, message string) error {
	config := nsq.NewConfig()
	producer, err := nsq.NewProducer(p.nsqdAddress, config)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("创建 NSQ 生产者失败: %v", err)
		return fmt.Errorf("创建 NSQ 生产者失败: %w", err)
	}
	defer producer.Stop()

	err = producer.Publish(topic, []byte(message))
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("向 NSQ 发布消息失败: %v", err)
		return fmt.Errorf("向 NSQ 发布消息失败: %w", err)
	}

	return nil
}

// Subscribe 订阅指定主题的消息
func (p *NSQPlugin) Subscribe(ctx context.Context, topic string, handler func(message string) error) error {
	config := nsq.NewConfig()
	consumer, err := nsq.NewConsumer(topic, "channel", config)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("创建 NSQ 消费者失败: %v", err)
		return fmt.Errorf("创建 NSQ 消费者失败: %w", err)
	}

	// 初始化或更新队列统计信息
	p.statsMutex.Lock()
	if _, exists := p.queueStats[topic]; !exists {
		p.queueStats[topic] = &QueueStats{
			TopicName: topic,
		}
	}
	stats := p.queueStats[topic]
	p.statsMutex.Unlock()

	consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
		// 更新统计信息
		p.statsMutex.Lock()
		stats.LastConsumedAt = time.Now().Unix()
		stats.ConsumedMessages++
		p.statsMutex.Unlock()

		err := handler(string(message.Body))
		if err != nil {
			// 更新失败统计
			p.statsMutex.Lock()
			stats.FailedMessages++
			p.statsMutex.Unlock()
		}
		return err
	}))

	err = consumer.ConnectToNSQD(p.nsqdAddress)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("连接到 NSQD 失败: %v", err)
		return fmt.Errorf("连接到 NSQD 失败: %w", err)
	}

	go func() {
		<-ctx.Done()
		consumer.Stop()
	}()

	// 将创建的消费者保存到activeConsumers中
	p.consumersMutex.Lock()
	p.activeConsumers[topic] = consumer
	p.consumersMutex.Unlock()
	return nil
}

// Close 关闭 NSQ 连接
func (p *NSQPlugin) Close() error {
	p.closed = true
	return nil
}

// GetStatus 获取NSQ的状态
func (p *NSQPlugin) GetStatus() MQStatus {
	// 尝试建立连接以检查连接状态
	config := nsq.NewConfig()
	producer, err := nsq.NewProducer(p.nsqdAddress, config)
	isConnected := err == nil
	if isConnected {
		producer.Stop() // 关闭测试连接
	} else {
		p.errorCount++
		p.lastError = err.Error()
	}

	return MQStatus{
		IsConnected:    isConnected,
		ConnectionInfo: p.nsqdAddress,
		StartTime:      p.startTime,
		ErrorCount:     p.errorCount,
		LastError:      p.lastError,
	}
}

// GetQueueStats 获取指定主题的队列统计信息
func (p *NSQPlugin) GetQueueStats(topic string) (QueueStats, error) {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	// 检查队列是否存在
	stats, exists := p.queueStats[topic]
	if !exists {
		return QueueStats{}, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}

	// NSQ 不提供简单的方式获取队列长度，这里只返回已有的统计信息
	return *stats, nil
}

// ListQueueStats 获取所有主题的队列统计信息
func (p *NSQPlugin) ListQueueStats() []QueueStats {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	result := make([]QueueStats, 0, len(p.queueStats))
	for _, stats := range p.queueStats {
		result = append(result, *stats)
	}

	return result
}

// GetPendingMessages 获取待处理的消息
func (p *NSQPlugin) GetPendingMessages(topic string, page, pageSize int) ([]string, error) {
	// 实现获取待处理消息的逻辑
	return []string{}, nil
}

// GetConsumedMessages 获取已消费的消息
func (p *NSQPlugin) GetConsumedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 实现获取已消费消息的逻辑
	return []MessageRecord{}, nil
}

// GetFailedMessages 获取失败的消息
func (p *NSQPlugin) GetFailedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 实现获取失败消息的逻辑
	return []MessageRecord{}, nil
}

// CloseQueue 关闭指定主题的队列
func (p *NSQPlugin) CloseQueue(topic string) error {
	// 检查连接状态
	if p.closed {
		return fmt.Errorf("NSQ 连接已关闭")
	}

	// 从活跃消费者中查找并关闭
	p.consumersMutex.Lock()
	defer p.consumersMutex.Unlock()

	consumer, exists := p.activeConsumers[topic]
	if !exists {
		// 如果没有活跃的消费者，只从统计信息中移除
		p.statsMutex.Lock()
		delete(p.queueStats, topic)
		p.statsMutex.Unlock()
		return nil
	}

	// 停止消费者
	consumer.Stop()

	// 从活跃消费者映射中移除
	delete(p.activeConsumers, topic)

	// 从统计信息中移除
	p.statsMutex.Lock()
	delete(p.queueStats, topic)
	p.statsMutex.Unlock()

	return nil
}

// ListTopics 获取所有主题列表
func (p *NSQPlugin) ListTopics() []string {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	topics := make([]string, 0, len(p.queueStats))
	for topic := range p.queueStats {
		topics = append(topics, topic)
	}

	return topics
}
